package com.itheima.delaytask.service.impl;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
import com.itheima.cache.CacheService;
import com.itheima.delaytask.constants.DelayTaskRedisKeyConstants;
import com.itheima.delaytask.constants.TaskStatusConstants;
import com.itheima.delaytask.constants.ZKdirConstants;
import com.itheima.delaytask.dto.Task;
import com.itheima.delaytask.exception.ScheduleSystemException;
import com.itheima.delaytask.exception.TaskNotExistException;
import com.itheima.delaytask.mapper.TaskInfoLogsMapper;
import com.itheima.delaytask.mapper.TaskInfoMapper;
import com.itheima.delaytask.po.TaskInfo;
import com.itheima.delaytask.po.TaskInfoLogs;
import com.itheima.delaytask.properties.SystemParamProperties;
import com.itheima.delaytask.service.TaskService;
import com.itheima.utils.ZKSelectMasterUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

import javax.annotation.PostConstruct;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Calendar;
import java.util.Date;
import java.util.List;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

/**
 * @program: virtual-trade-parent
 * @description: 延迟任务的相关的service实现, 主要是用来处理任务的
 * @author: zhanghz001
 * @create: 2021-07-09 09:30
 **/
// @Service
@Slf4j
public class ZhzTaskServiceImpl implements TaskService {
    
    @Autowired
    private TaskInfoMapper taskInfoMapper;
    
    @Autowired
    private TaskInfoLogsMapper taskInfoLogsMapper;
    
    @Autowired
    private CacheService cacheService;
    
    @Autowired
    private ThreadPoolTaskExecutor threadPoolTaskExecutor;
    
    @Autowired
    private SystemParamProperties paramProperties;
    @Autowired
    private ThreadPoolTaskScheduler threadPoolTaskScheduler;
    @Autowired
    private ZKSelectMasterUtil zkSelectMasterUtil;
    
    //加载到zset的数据的截止时间,比如未来2分钟内,
    private long nextScheduleTime;
    
    @PostConstruct
    public void syncData() {
        //启动后去争抢注册成为主节点
        zkSelectMasterUtil.setZkAddress(ZKdirConstants.schedule_leaderPath);
        
        try {
            TimeUnit.SECONDS.sleep(1);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            
        }
        threadPoolTaskScheduler.scheduleAtFixedRate(
                new Runnable() {
                    @Override
                    public void run() {
                        if (zkSelectMasterUtil.checkMaster(ZKdirConstants.schedule_leaderPath)) {
                            
                            log.info("{}主节点,开始进行定时数据恢复",
                                    LocalDateTime.now().format(DateTimeFormatter.ISO_DATE_TIME));
                            loadData();
                        } else {
                            log.info("从节点备份");
                        }
                    }
                },
                TimeUnit.MINUTES.toMillis(paramProperties.getPreLoad())
        );
    }
    
    public void loadData() {
        long start = System.currentTimeMillis();
        
        //先清除缓存中的所有数据
        clearCache();
        
        //查询任务表中的所有分组
        QueryWrapper<TaskInfo> groupWrapper = new QueryWrapper<>();
        groupWrapper.lambda()
                .select(TaskInfo::getTaskType, TaskInfo::getPriority)
                .groupBy(TaskInfo::getTaskType, TaskInfo::getPriority);
        List<TaskInfo> taskInfoGroups = taskInfoMapper.selectList(groupWrapper);
        
        //每组数据加到缓存
        if (CollectionUtils.isEmpty(taskInfoGroups)) {
            return;
        }
        
        //获取未来两分钟的时间点,存到redis
        Calendar calendar = Calendar.getInstance();
        calendar.add(Calendar.MINUTE, paramProperties.getPreLoad());
        nextScheduleTime = calendar.getTimeInMillis();
        
        //存入redis
        cacheService.set(DelayTaskRedisKeyConstants.nextScheduleTime, nextScheduleTime + "");
        //
        CountDownLatch countDownLatch = new CountDownLatch(taskInfoGroups.size());
        for (TaskInfo taskInfoGroup : taskInfoGroups) {
            threadPoolTaskExecutor.submit(new Callable<Void>() {
                @Override
                public Void call() throws Exception {
                    //查询每个分组
                    QueryWrapper<TaskInfo> queryWrapper = new QueryWrapper<>();
                    queryWrapper.lambda()
                            .eq(TaskInfo::getTaskType, taskInfoGroup.getTaskType())
                            .eq(TaskInfo::getPriority, taskInfoGroup.getPriority())
                    ;
                    List<TaskInfo> taskInfos = taskInfoMapper.selectList(queryWrapper);
                    
                    //调用方法添加到缓存 
                    for (TaskInfo taskInfo : taskInfos) {
                        Task task = new Task();
                        BeanUtils.copyProperties(taskInfo, task);
                        task.setExecuteTime(taskInfo.getExecuteTime().getTime());
                        addTaskToCache(task);
                    }
                    countDownLatch.countDown();
                    return null;
                }
            });
            
            //
            try {
                countDownLatch.await(5, TimeUnit.MINUTES);
                log.info("多线程分组数据恢复耗时:{}", System.currentTimeMillis() - start);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
    
    private void clearCache() {
        //清除未来future_ 开头的未来消费队列和清除 topic_ 需要马上消费的队列
        //对于模糊的查询都用scan进行替代
        Set<String> topicKeys = cacheService.scan(DelayTaskRedisKeyConstants.TOPIC + "*");
        Set<String> futureKeys = cacheService.scan(DelayTaskRedisKeyConstants.FUTURE + "*");
        cacheService.delete(topicKeys);
        cacheService.delete(futureKeys);
    }
    
    @Override
    @Transactional(rollbackFor = RuntimeException.class)
    public long addTask(Task task) throws ScheduleSystemException {
        //先将任务添加到数据库
        boolean result = addTaskToDb(task);
        if (result) {
            //添加成功后，将任务添加到缓存
            addTaskToCache(task);
        }
        return task.getTaskId();
    }
    
    /**
     * Task updateDb(taskId);删除任务表数据，更新任务日志表状态为已取消
     * void removeTaskFromCache(Task task),从redis中删除任务数据
     *
     * @param taskId 任务id
     * @return
     * @throws TaskNotExistException
     * @throws ScheduleSystemException
     */
    @Override
    @Transactional
    public boolean cancelTask(long taskId) throws TaskNotExistException, ScheduleSystemException {
        boolean success = false;
        
        //去删除db中的任务
        Task task = updateDb(taskId, TaskStatusConstants.CANCELLED);
        if (task != null) {
            removeTaskFromCache(task);
            success = true;
        }
        
        return success;
    }
    
    /**
     * 消费队列接口,可能获取多个,但是只消费一个
     *
     * @return
     * @throws ScheduleSystemException
     */
    @Override
    @Transactional(rollbackFor = RuntimeException.class)
    public Task poll() throws ScheduleSystemException {
        Task task = null;
        
        //从zset中拉取当前需要执行的任务,按照分数获取,分数就是任务执行的时间
        Set<String> taskStringSet = cacheService.zRangeByScore(
                DelayTaskRedisKeyConstants.DBCACHE,
                0, System.currentTimeMillis());
        if (CollectionUtils.isEmpty(taskStringSet)) {
            return null;
        }
        try {
            //获取第一个,进行消费
            String taskString = taskStringSet.iterator().next();
            task = JSON.parseObject(taskString, Task.class);
            
            //更新到数据库任务信息--删除任务数据,日志状态改成已经执行--复用之前的方法
            updateDb(task.getTaskId(), TaskStatusConstants.EXECUTED);
            
            //删除缓存中该任务数据
            cacheService.zRemove(DelayTaskRedisKeyConstants.DBCACHE, taskString);
        } catch (Exception e) {
            log.error("poll task exception ,msg={}", e.getMessage());
            throw new ScheduleSystemException(e);
        }
        
        //
        return task;
    }
    
    @Override
    @Transactional
    public Task poll(int type, int priority) throws ScheduleSystemException {
        Task task = null;
        
        //构造对应的分组key
        String key = type + "_" + priority;
        
        //获取相关的任务
        String taskJson = cacheService.lRightPop(DelayTaskRedisKeyConstants.TOPIC + key);
        //说明没有这个key
        if (StringUtils.isEmpty(taskJson)) {
            return null;
        }
        try {
            
            task = JSON.parseObject(taskJson, Task.class);
        } catch (Exception e) {
            log.error("poll task exception,type={},priority={}", task, priority);
            throw new ScheduleSystemException(e);
        }
        
        //
        return task;
    }
    
    @Override
    // @Scheduled(cron = "*/1 * * * * ?")
    public void refresh() {
        threadPoolTaskExecutor.execute(new Runnable() {
            @Override
            public void run() {
                log.info("{}进行了定时刷新", LocalDateTime.now().format(DateTimeFormatter.ISO_LOCAL_DATE_TIME));
                try {
                    TimeUnit.SECONDS.sleep(2);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                //返回未来数据集合zset中所有的key,是否要放到list中去
                Set<String> futureKeySet = cacheService.scan(DelayTaskRedisKeyConstants.FUTURE + "*");
                if (CollectionUtils.isEmpty(futureKeySet)) {
                    return;
                }
                for (String futureKey : futureKeySet) {
                    //根据每一个key从当前分组下,获取需要执行的任务set集合 
                    Set<String> values = cacheService.zRangeByScore(futureKey, 0, System.currentTimeMillis());
                    if (CollectionUtils.isEmpty(values)) {
                        continue;
                    }
                    //将这组数据添加到消费者队列的对应分组上,并从zset中删除
                    String key = futureKey.split(DelayTaskRedisKeyConstants.FUTURE)[1];//future_  2001_1
                    // for (String taskJson : values) {
                    //     cacheService.lLeftPush(DelayTaskRedisKeyConstants.TOPIC + key, taskJson);
                    //     cacheService.zRemove(futureKey, taskJson);
                    // }
                    //使用管道技术进行优化
                    cacheService.refreshWithPipeline(futureKey, DelayTaskRedisKeyConstants.TOPIC + key, values);
                }
                
            }
        });
        
    }
    
    /**
     * Task updateDb(taskId);删除任务表数据，更新任务日志表状态为已取消
     *
     * @param taskId
     * @param status
     * @return
     */
    private Task updateDb(long taskId, int status) {
        Task task = null;
        
        //修改任务日志表中的状态,
        TaskInfoLogs taskInfoLogs = taskInfoLogsMapper.selectById(taskId);
        if (taskInfoLogs == null) {
            throw new TaskNotExistException("task not exit ,taskid={}", taskId);
        }
        
        try {
            //更新日志数据
            UpdateWrapper<TaskInfoLogs> updateWrapper = new UpdateWrapper<>();
            updateWrapper.lambda().set(TaskInfoLogs::getStatus, status)
                    .eq(TaskInfoLogs::getTaskId, taskInfoLogs.getTaskId());
            taskInfoLogsMapper.update(null, updateWrapper);
            
            //删除任务数据
            taskInfoMapper.deleteById(taskId);
            
            //构造返回值
            task = new Task();
            //封装数据
            BeanUtils.copyProperties(taskInfoLogs, task);
            task.setExecuteTime(taskInfoLogs.getExecuteTime().getTime());
            
        } catch (Exception e) {
            log.error("cancel task exception ,taskId={}", taskId);
            throw new ScheduleSystemException(e);
        } finally {
            
        }
        return task;
    }
    
    /**
     * 从redis中删除任务数据
     *
     * @param task
     */
    private void removeTaskFromCache(Task task) {
        String key = task.getTaskType() + "_" + task.getPriority();
        //判断
        if (task.getExecuteTime() < System.currentTimeMillis()) {
            //从消费者队列中删除
            cacheService.lRemove(DelayTaskRedisKeyConstants.TOPIC + key, 0, JSON.toJSONString(task));
        } else {
            //从zset中删除
            cacheService.zRemove(DelayTaskRedisKeyConstants.FUTURE + key, JSON.toJSONString(task));
        }
        
        //暂时这么写,后续优化
        // cacheService.zRemove(DelayTaskRedisKeyConstants.DBCACHE, JSON.toJSONString(task));
    }
    
    /**
     * 先将任务添加到数据库
     *
     * @param task
     * @return
     */
    private boolean addTaskToDb(Task task) {
        boolean success = false;
        
        //
        try {
            //需要执行的任务需要先进入数据库
            
            TaskInfo taskInfo = new TaskInfo();
            taskInfo.setTaskType(task.getTaskType());
            taskInfo.setPriority(task.getPriority());
            taskInfo.setParameters(task.getParameters());
            taskInfo.setExecuteTime(new Date(task.getExecuteTime()));
            
            taskInfoMapper.insert(taskInfo);
            
            //填充返回的主键值-任务id
            task.setTaskId(taskInfo.getTaskId());
            
            //保存任务日志信息---任务日志和任务字段差不多，多了一个版本号和任务状态
            TaskInfoLogs taskInfoLogs = new TaskInfoLogs();
            //两个对象进行属性值传递，属性少建议用settter，速度最快，属性多用setter不优雅，可以使用属性
            // 拷贝工具，譬如：cglib Beancopier，Spring beanUtils，Apache beanUtils，中都提供了类似的方法
            
            BeanUtils.copyProperties(taskInfo, taskInfoLogs);
            taskInfoLogs.setVersion(1);
            //状态为刚提交的状态
            taskInfoLogs.setStatus(TaskStatusConstants.SCHEDULED);
        } catch (Exception e) {
            log.error("add task exception ,task={}", task);
            throw new ScheduleSystemException(e);
        }
        
        //返回是成功
        return success;
        
    }
    
    /**
     * 添加成功后，将任务添加到缓存
     *
     * @param task
     */
    private void addTaskToCache(Task task) {
        //进行分组,根据类型和优先级进行分组
        String key = task.getTaskType() + "_" + task.getPriority();
        
        //从redis获取未来的时间节点,如果redis没有就自己造一个
        long nextScheduleTime = fetchNextScheduleTime();
        //根据时间判断
        if (task.getExecuteTime() <= System.currentTimeMillis()) {
            //如果已经是需要被执行了,就放到topic_的list队列中
            cacheService.lLeftPush(DelayTaskRedisKeyConstants.TOPIC + key,
                    JSON.toJSONString(task));
            
        } else if (task.getExecuteTime() <= nextScheduleTime) {
            
            //任务到zset中排队等待        
            cacheService.zAdd(DelayTaskRedisKeyConstants.FUTURE + key,
                    JSON.toJSONString(task),
                    task.getExecuteTime());
            
        }
        
        //暂时存储到zset,后续会对此进行优化
        // cacheService.zAdd(DelayTaskRedisKeyConstants.DBCACHE,
        //         JSON.toJSONString(task),
        //         task.getExecuteTime());
    }
    
    private long fetchNextScheduleTime() {
        String nextScheduleTimeString = cacheService.get(DelayTaskRedisKeyConstants.nextScheduleTime);
        if (StringUtils.isEmpty(nextScheduleTimeString)) {
            //自己造一个下次执行时间
            
            //获取未来两分钟的时间点,存到redis
            Calendar calendar = Calendar.getInstance();
            calendar.add(Calendar.MINUTE, paramProperties.getPreLoad());
            this.nextScheduleTime = calendar.getTimeInMillis();
            
            //存入redis
            cacheService.set(DelayTaskRedisKeyConstants.nextScheduleTime, nextScheduleTime + "");
            
            log.info("缓存中没有找到,自己构造了 nextScheduleTime :{}", nextScheduleTimeString);
            return nextScheduleTime;
        } else {
            log.info("从缓存中获取到了nextScheduleTime :{}", nextScheduleTimeString);
            return Long.parseLong(nextScheduleTimeString);
        }
    }
}
